package com.mall.promo.mq;

import com.alibaba.fastjson.JSON;
import com.mall.order.dto.CreateSeckillOrderRequest;
import com.mall.promo.cache.CacheManager;
import com.mall.promo.dal.persistence.PromoItemMapper;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.xml.stream.events.StartDocument;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

/**
 * Author  kkw
 * Date  2021/6/25 11:14
 * Version 1.0
 * Description :分布式事务生产者
 */
@Component
public class PromoOrderProducer {

    @Autowired
    PromoItemMapper promoItemMapper;
    @Autowired
    CacheManager cacheManager;
    @Autowired
    RedissonClient redissonClient;

    private TransactionMQProducer producer;

    @Value("${mq.nameserver.address}")
    private String nameSrvAddress;
    @Value("${mq.topicname}")
    private String topicName;

    @PostConstruct
    public void init() {
        producer = new TransactionMQProducer("promo_order_producer");

        producer.setNamesrvAddr(nameSrvAddress);
        //设置监听器,监听consumer返回的消息
        producer.setTransactionListener(new TransactionListener() {
            //本地事务扣减库存
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {

                System.out.println(message);
                Map<String,Long> map= (Map<String, Long>) o;
                Long productId = map.get("productId");
                Long psId = map.get("psId");
                //把执行结果保存起来,防止网络通讯失败
                //引入redis,把执行结果暂存在redis里面,此时需要一个唯一的key值
                String transactionId = message.getTransactionId();
                //做的长一点,确保其唯一性
                String transactionKey="promo_order_transaction"+transactionId;

                //redis没实现分布式锁
                //redissonClient实现了分布式锁
                //获取一把锁,只针对秒杀场次的某个商品
                String lockTransactionKey="promo_item_stock:"+psId+"_"+productId;
                RLock lock = redissonClient.getLock(lockTransactionKey);
                lock.lock();
                try {
                    //在这里扣减库存
                    Integer effectRow = promoItemMapper.decreaseStock(productId, psId);
//                    System.out.println(effectRow);
                    if (effectRow < 1) {
                        //扣减库存失败
                        cacheManager.setCache(transactionKey,"fail",1);
                        //expire表示,一天之后过期
                        return LocalTransactionState.ROLLBACK_MESSAGE;
                    }
                }finally {
                    lock.unlock();
                }

                //本地事务执行成功
                cacheManager.setCache(transactionKey,"success",1);
                String s = cacheManager.checkCache(transactionKey);
                System.out.println(s);
                return LocalTransactionState.COMMIT_MESSAGE;
            }

            //扣架库存验证,回查
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                //查询本地事务的结果
                //做的长一点,确保其唯一性,和上面方法的key值保持一致,才能从redis里面获取到消息
                String transactionKey="promo_order_transaction"+messageExt.getTransactionId();
                String cacheResult = cacheManager.checkCache(transactionKey);
                if (cacheResult == null || cacheResult.trim().isEmpty()) {
                    return LocalTransactionState.UNKNOW;
                }
                if ("success".equals(cacheResult)) {
                    return LocalTransactionState.COMMIT_MESSAGE;
                }
                return LocalTransactionState.ROLLBACK_MESSAGE;
            }
        });
        try {
            producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    /**
     * 用来发送,秒杀下单的事务方法
     * @param request
     *request在orderService里面
     * @return
     */
    public boolean sendPromoOrderMessage(CreateSeckillOrderRequest request,Long productId,Long psId) {

//        把需要的参数封装到Map里面
        HashMap<String, Long> map = new HashMap<>();
        map.put("productId",productId);
        map.put("psId",psId);
        //request转化为json字符串消息
        String secKillOrder = JSON.toJSONString(request);
        Message message = new Message();
        message.setTopic(topicName);
        //订单事务的数据
        message.setBody(secKillOrder.getBytes(StandardCharsets.UTF_8));
        TransactionSendResult sendResult=null;
        //需要传递参数,arg先定为null
        //可以直接使用arg传递使用的消息,比使用message方便点
        try {
            sendResult = producer.sendMessageInTransaction(message, map);
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        //获取事务消息对应本地事务的执行结果
        if (sendResult != null && LocalTransactionState.COMMIT_MESSAGE.equals(sendResult.getLocalTransactionState())) {
            return true;
        }
        return false;
    }
}
